use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use async_trait::async_trait;
use tokio::time::sleep;
use tracing::Level;

use protoactor_rs::actor::{Actor, ActorContext, ActorError, PID, Props};
use protoactor_rs::pubsub::{PubSub, PubSubBatch, Publisher, Subscriber};
use protoactor_rs::system::ActorSystem;

// 示例消息
#[derive(Debug, Clone)]
struct WeatherUpdate {
    location: String,
    temperature: f32,
    humidity: f32,
    timestamp: chrono::DateTime<chrono::Utc>,
}

// 天气订阅者 Actor
struct WeatherSubscriber {
    name: String,
}

#[async_trait]
impl Actor for WeatherSubscriber {
    type Context = protoactor_rs::actor::DefaultActorContext;
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn std::any::Any + Send>) -> Result<(), ActorError> {
        if let Some(batch) = msg.downcast_ref::<PubSubBatch>() {
            if let Some(update) = batch.message.downcast_ref::<WeatherUpdate>() {
                println!("[{}] Received weather update for {}: {}°C, {}% humidity", 
                    self.name, update.location, update.temperature, update.humidity);
            }
        }
        Ok(())
    }
    
    async fn started(&self, _ctx: &mut Self::Context) -> Result<(), ActorError> {
        println!("[{}] Weather subscriber started", self.name);
        Ok(())
    }
}

// 天气发布者 Actor
struct WeatherPublisher {
    publisher: Option<Publisher>,
    locations: Vec<String>,
    interval: Duration,
}

#[async_trait]
impl Actor for WeatherPublisher {
    type Context = protoactor_rs::actor::DefaultActorContext;
    
    async fn receive(&self, ctx: &mut Self::Context, msg: Box<dyn std::any::Any + Send>) -> Result<(), ActorError> {
        if let Some(_) = msg.downcast_ref::<PublishWeather>() {
            if let Some(publisher) = &self.publisher {
                // 生成随机天气数据
                for location in &self.locations {
                    let update = WeatherUpdate {
                        location: location.clone(),
                        temperature: rand::random::<f32>() * 30.0,  // 0-30°C
                        humidity: rand::random::<f32>() * 100.0,    // 0-100%
                        timestamp: chrono::Utc::now(),
                    };
                    
                    if let Err(e) = publisher.publish(update).await {
                        println!("Failed to publish weather update: {}", e);
                    }
                }
                
                // 安排下一次更新
                ctx.schedule_once(self.interval, move |ctx| {
                    ctx.send(ctx.self_pid(), PublishWeather {}).unwrap();
                });
            }
        } else if let Some(init) = msg.downcast_ref::<InitPublisher>() {
            let mut this = self.clone();
            this.publisher = Some(init.publisher.clone());
            println!("Weather publisher initialized for locations: {:?}", self.locations);
            
            // 开始发布
            ctx.send(ctx.self_pid(), PublishWeather {}).unwrap();
        }
        
        Ok(())
    }
}

impl Clone for WeatherPublisher {
    fn clone(&self) -> Self {
        Self {
            publisher: self.publisher.clone(),
            locations: self.locations.clone(),
            interval: self.interval,
        }
    }
}

// 初始化发布者消息
struct InitPublisher {
    publisher: Publisher,
}

// 发布天气消息
struct PublishWeather {}

#[tokio::main]
async fn main() -> Result<()> {
    // 初始化日志
    tracing_subscriber::fmt()
        .with_max_level(Level::DEBUG)
        .init();
    
    // 创建 Actor 系统
    let system = Arc::new(ActorSystem::new()?);
    
    // 创建 PubSub 系统
    let pubsub = PubSub::new(system.clone())
        .with_subscriber_timeout(Duration::from_secs(3));
    
    // 启动 PubSub 系统
    pubsub.start().await?;
    
    // 创建天气订阅者
    let subscriber1_props = Props::new(Box::new(|| {
        Box::new(WeatherSubscriber {
            name: "Tokyo Station".to_string(),
        })
    }));
    
    let subscriber2_props = Props::new(Box::new(|| {
        Box::new(WeatherSubscriber {
            name: "New York Station".to_string(),
        })
    }));
    
    let subscriber1_pid = system.spawn(subscriber1_props).await?;
    let subscriber2_pid = system.spawn(subscriber2_props).await?;
    
    // 创建订阅者对象
    let weather_topic = "weather.updates";
    let subscriber1 = Subscriber::new(pubsub.clone(), weather_topic.to_string(), subscriber1_pid.clone());
    let subscriber2 = Subscriber::new(pubsub.clone(), weather_topic.to_string(), subscriber2_pid.clone());
    
    // 订阅主题
    subscriber1.subscribe().await?;
    subscriber2.subscribe().await?;
    
    println!("Subscribers registered to 'weather.updates' topic");
    
    // 创建发布者 Actor
    let publisher_props = Props::new(Box::new(|| {
        Box::new(WeatherPublisher {
            publisher: None,
            locations: vec![
                "Tokyo".to_string(), 
                "New York".to_string(), 
                "London".to_string(), 
                "Sydney".to_string(),
                "Beijing".to_string(),
            ],
            interval: Duration::from_secs(3),
        })
    }));
    
    let publisher_pid = system.spawn(publisher_props).await?;
    
    // 创建发布者对象并初始化 Actor
    let publisher = Publisher::new(pubsub.clone(), weather_topic.to_string());
    let init_message = InitPublisher {
        publisher: publisher.clone(),
    };
    
    system.send(&publisher_pid, init_message).await?;
    
    // 运行一段时间
    println!("Publishing weather updates for 15 seconds...");
    sleep(Duration::from_secs(15)).await;
    
    // 取消订阅并停止 Actors
    println!("Unsubscribing and stopping actors...");
    subscriber1.unsubscribe().await?;
    subscriber2.unsubscribe().await?;
    
    system.stop(&subscriber1_pid).await?;
    system.stop(&subscriber2_pid).await?;
    system.stop(&publisher_pid).await?;
    
    println!("Example completed");
    Ok(())
} 